package com.zqx.kafkaproject.kafkademo.components;

import com.zqx.kafkaproject.kafkademo.entity.DayEntrustDomain;
import com.zqx.kafkaproject.kafkademo.repository.IDayEntrustDao;
import com.zqx.kafkaproject.kafkademo.utils.KeyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.servlet.ServletContextEvent;
import java.util.Optional;

@Component
@Slf4j
public class KafkaReceiver {

    @Autowired
    private  IDayEntrustDao aIDayEntrustDao;
    @KafkaListener(topics = {"zhisheng"})
    public void listen(ConsumerRecord<?,?> record)
    {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());

        if(kafkaMessage.isPresent())
        {
            Object message = kafkaMessage.get();
            log.info("【kafka 的record】{}",record);
            log.info("【kafka 的message】{}",message);
            DayEntrustDomain dayEntrustDomain = new DayEntrustDomain();
            dayEntrustDomain.setId(KeyUtil.genUniqueKey());
            dayEntrustDomain.setMessage((String) message);
            aIDayEntrustDao.save(dayEntrustDomain);
        }
    }

    public void pub(){

    }

    /*@KafkaListener(topics = {"zhisheng"})
    public void listen(ServletContextEvent sce)
    {
        Thread t = new Thread(new KafkaConsumerRunnable(sce));
        t.start();
    }*/

}
